package com.isyscore.os.metadata.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.isyscore.os.core.exception.DataFactoryException;
import com.isyscore.os.core.exception.ErrorCode;
import com.isyscore.os.core.mapper.DataFlowTaskMapper;
import com.isyscore.os.core.model.entity.DataFlowDefinition;
import com.isyscore.os.core.model.entity.DataFlowTask;
import com.isyscore.os.metadata.controller.OpenApiController;
import com.isyscore.os.metadata.enums.KettleJobStatus;
import com.isyscore.os.metadata.enums.KettleJobTriggerType;
import com.isyscore.os.metadata.service.executor.FlowExecutor;
import com.isyscore.os.metadata.kettle.base.FlowMetaCodec;
import com.isyscore.os.metadata.kettle.dto.FlowTaskDTO;
import com.isyscore.os.metadata.kettle.dto.FlowTaskPageReqDTO;
import com.isyscore.os.metadata.service.DataFlowDefinitionService;
import com.isyscore.os.metadata.service.DataFlowTaskService;
import lombok.RequiredArgsConstructor;
import org.pentaho.di.trans.TransMeta;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Collectors;

@Service
@RequiredArgsConstructor
public class DataFlowTaskServiceImpl extends ServiceImpl<DataFlowTaskMapper, DataFlowTask> implements DataFlowTaskService {
    @Autowired
    private DataFlowDefinitionService dataFlowDefinitionService;

    @Autowired
    private FlowMetaCodec codec;


    @Override
    public FlowTaskDTO getFlowTask(String id) {
        DataFlowTask dataFlowTask = this.getOne(Wrappers.lambdaQuery(DataFlowTask.class).eq(DataFlowTask::getId, id), false);
        if (ObjectUtil.isNotNull(dataFlowTask)) {
            FlowTaskDTO flowTaskDTO = new FlowTaskDTO();
            DataFlowDefinition dataFlowDefinition = dataFlowDefinitionService.getById(dataFlowTask.getDefinitionId());
            flowTaskDTO.valueOf(dataFlowTask, dataFlowDefinition.getName());
            return flowTaskDTO;
        } else {
            throw new DataFactoryException(ErrorCode.TASK_ID_NOT_FOUND, id);
        }

    }

    @Override
    public String runTask(String definitionId, Map<String, String> params, KettleJobTriggerType triggerType, OpenApiController.CallBack callBack) {
        if (params == null) {
            params = new HashMap<>();
        }

        FlowExecutor transExecutor;
        String xml = dataFlowDefinitionService.getById(definitionId).getKettleDefinition();
        String executionId;
        TransMeta transMeta = (TransMeta) codec.decode(xml);
        transExecutor = FlowExecutor.createFlowExecutor(params, transMeta, definitionId, triggerType, callBack);
        transExecutor.execute(false);
        executionId = transExecutor.getExecutionId();
        return executionId;
    }

    @Override
    public void stopTask(String taskId) {
        FlowExecutor executor = FlowExecutor.getExecutors().getIfPresent(taskId);
        if (executor != null) {
            executor.stopAndRemove();
        }
    }

    @Override
    public IPage<FlowTaskDTO> page(FlowTaskPageReqDTO req) {
        LambdaQueryWrapper<DataFlowDefinition> dataFlowDefinitionQuery = Wrappers.lambdaQuery(DataFlowDefinition.class)
                .like(StrUtil.isNotBlank(req.getFlowName()), DataFlowDefinition::getName, req.getFlowName())
                .eq(StrUtil.isNotBlank(req.getDefinitionId()), DataFlowDefinition::getId, req.getDefinitionId());
        List<DataFlowDefinition> dataFlowDefinitionList = dataFlowDefinitionService.list(dataFlowDefinitionQuery);
        if (CollUtil.isNotEmpty(dataFlowDefinitionList)) {
            LambdaQueryWrapper<DataFlowTask> dataFlowTaskLambdaQuery = Wrappers.lambdaQuery(DataFlowTask.class)
                    .in((StrUtil.isNotBlank(req.getFlowName()) || StrUtil.isNotBlank(req.getDefinitionId())), DataFlowTask::getDefinitionId, dataFlowDefinitionList.stream().map(DataFlowDefinition::getId).collect(Collectors.toList()))
                    .ge(ObjectUtil.isNotNull(req.getStartTime()), DataFlowTask::getStartTime, req.getStartTime())
                    .le(ObjectUtil.isNotNull(req.getEndTime()), DataFlowTask::getStartTime, req.getEndTime())
                    .eq(ObjectUtil.isNotNull(req.getStatus()), DataFlowTask::getStatus, req.getStatus());
            if (CollUtil.isEmpty(req.getOrders())) {
                dataFlowTaskLambdaQuery.orderByDesc(DataFlowTask::getStartTime);
            }
            //用map存储定义ID和定义名称
            Map<Long, String> finalFlowMap = dataFlowDefinitionList.stream().collect(Collectors.toMap(DataFlowDefinition::getId, DataFlowDefinition::getName));
            return this.page(req.toPage(), dataFlowTaskLambdaQuery).convert(dataFlowTask -> {
                FlowTaskDTO flowTaskDTO = new FlowTaskDTO();
                flowTaskDTO.valueOf(dataFlowTask, finalFlowMap.get(dataFlowTask.getDefinitionId()));
                return flowTaskDTO;
            });
        } else {
            return new Page<>();
        }
    }

    @Override
    public List<String> runningFlows() {
        return new ArrayList<>(FlowExecutor.getRunningFlow());
    }

    @Override
    public String taskStatus(String taskId) {
        FlowTaskDTO flowTask = this.getFlowTask(taskId);
        return KettleJobStatus.getByCode(flowTask.getStatus()).name();
    }


}
